package com.lzq.chatroom_project.common.config.redis_caffeine_config;

import com.github.benmanes.caffeine.cache.Cache;
import io.lettuce.core.RedisClient;
import io.lettuce.core.TrackingArgs;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.codec.StringCodec;
import io.lettuce.core.support.caching.CacheFrontend;
import io.lettuce.core.support.caching.ClientSideCaching;
import lombok.Getter;
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;
import java.util.Map;
import java.util.Set;

public class CacheFrontendContext {
    private static final Logger log = LoggerFactory.getLogger(CacheFrontendContext.class);
    @Getter
    private CacheFrontend cacheFrontend;
    private RedisClient redisClient;
    @Getter
    private Cache cache;
    private Cache waitingSyncDataCache;
    @Autowired
    private TryConnectTime tryConnectTime;
    @Autowired
    private RedisTemplate redisTemplate;
    StatefulRedisConnection<String, String> connection;

    public CacheFrontendContext(RedisClient redisClient, Cache cache, Cache waitingSyncDataCache) {
        this.redisClient = redisClient;
        this.cache = cache;
        this.waitingSyncDataCache = waitingSyncDataCache;
    }

    @SneakyThrows
    public void check() {
        if (connection != null) {
            if (connection.isOpen()) {
                if (!waitingSyncDataCache.asMap().entrySet().isEmpty()) {
                    syncDataToRedis(waitingSyncDataCache);
                }
                tryConnectTime.reset();
                return;
            }
        }
        try {
            tryConnectTime.increase();
            if (tryConnectTime.isDegrade()) Thread.sleep(5000);
            //重新建立连接
            connection = redisClient.connect();
            CacheFrontend<String, String> frontend = ClientSideCaching.enable(
                    new CaffeineCacheAccessor(cache),
                    connection,
                    TrackingArgs.Builder.enabled()
            );
            this.cacheFrontend = frontend;
            if (!waitingSyncDataCache.asMap().entrySet().isEmpty()) {
                syncDataToRedis(waitingSyncDataCache);
            }
            //添加监听,如果redis数据变动,caffeine自动清除缓存
            connection.addListener(message -> {
                List<Object> content = message.getContent(StringCodec.UTF8::decodeKey);
                log.info("type:{}, content:{}",message.getType(), content);
                if (message.getType().equalsIgnoreCase("invalidate")) {
                    List<String> keys = (List<String>) content.get(1);
                    keys.forEach(key -> {
                        cache.invalidate(key);
                    });
                }
            });
            log.warn("The redis client side connection had been reconnected.");
        }catch (RuntimeException e) {
            log.error("The redis client side connection had been disconnected, waiting reconnect...", e);
        }
    }

    private void syncDataToRedis(Cache waitingSyncDataCache) {
        Set<Map.Entry<String, String>> entrySet = waitingSyncDataCache.asMap().entrySet();
        for (Map.Entry<String, String> entry : entrySet) {
            if (!redisTemplate.hasKey(entry.getKey())) {
                redisTemplate.opsForValue().set(entry.getKey(), entry.getValue());
                log.info("同步key:{},value:{}到Redis客户端",entry.getKey(), entry.getValue());
            }
            waitingSyncDataCache.invalidate(entry.getKey());
        }
    }
}
